package com.atguigu.partitioner;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 自定义分区器
 *
 * @author code1997
 */
@Slf4j
public class MyPartitioner implements Partitioner {
    /**
     * 根据接受的参数，我们可以理解到，当到了分区器的时候，消息已经被序列化过了
     * 分区的策略：
     * 如果指定分区，直接发送给指定的分区，就不走分区器
     * 如果存在key，对key求hashcode，然后对可用分区数求余。
     * 如果key为null，使用round-robin算法。
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //获取当前topic的可用分区数
        int partitionCount = cluster.partitionCountForTopic(topic);
        log.info("topic:{},当前可用的分区数为:{}", topic, partitionCount);
        return 1;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
